Thread Pool

A thread pool is an array of threads. Each thread can be managed individually or all the threads in the pool can be managed together. During program execution, all threads in the pool can begin to work together on a specific task. Then, the program waits for all the threads to complete and resume execution with one single thread.

Un arreglo de threads es un conjunto de threads. Cada thread puede ser administrada individualmente o todas las threads en el arreglo pueden ser administradas juntas. Durante la ejecución del programa, todas las threads in el arreglo pueden empezar a trabajar juntas en una tarea específica. Entonces, el programa espera a que todas las threads terminen y continua la ejecución con una sola thread.

Problem 1

Create a Wintempla Dialog application called Together to test a thread pool. After creating the application, insert a multi-line textbox called tbxOutput.

Cree una aplicación de diálogo llamado Together para probar un arreglo de threads. Después de crear el programa, inserte una caja de texto de multi-línea llamada tbxOutput.

Step A

Edit the Together.h file. Do not forget to implement the Mt::IThreadX interface.

Edite el archivo Together.h. No sé olvide de implementar la interface Mt::IThreadX.

Together.h

#pragma once //______________________________________ Together.h
#include "Resource.h"
class Together: public Win::Dialog, public Mt::IThreadX
{
public:
     Together()
     {
     }
     ~Together()
     {
     }
     wstring text;
     Mt::CriticalSection cs;
     Mt::ThreadPool pool;
     //_____________________________________________________________ Mt::IThreadX
     DWORD ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads);
protected:
     . . .
};

Step B

Edit the Together.cpp file.

Edite el archivo Together.cpp.

Together.cpp

. . .
void Together::Window_Open(Win::Event& e)
{
     //________________________________________________________ 1. Begin all threads in the pool
     pool.BeginAll(*this);
     //________________________________________________________ 2. Wait for all threads in the pool to finish
     pool.WaitForAll();
     tbxOutput.Text = text;
}

DWORD Together::ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads)
{
     wchar_t tmp[256];
     _snwprintf_s(tmp, 256, _TRUNCATE, L"threadIndex=%d\t\tnumThreads=%d\r\n", (int)threadIndex, (int)numThreads);
     cs.Enter();
     text += tmp;
     cs.Leave();
     return 0;
}

Tip

In a thread pool, the execution duration of each thread in the pool is different. Therefore, the threads in the pool finish at different times.

En un arreglo de threads, la duración de ejecución para cada thread in el arreglo es diferente. Por lo tanto, las threads en el arreglo terminan en tiempos diferente.

Problem 2

Create a Wintempla Dialog application called Distribution to test how to distribute work using a thread pool. After creating the application, insert a multi-line textbox called tbxOutput.

Cree una aplicación de diálogo llamado Distribution para probar cómo distribuir trabajo usando un conjunto de threads. Después de crear el programa, inserte una caja de texto de multi-línea llamada tbxOutput.

Step A

Edit the Distribution .h file. Do not forget to implement the Mt::IThreadX interface.

Edite el archivo Distribution .h. No sé olvide de implementar la interface Mt::IThreadX.

Distribution .h

#pragma once //______________________________________ Distribution.h
#include "Resource.h"
class Distribution: public Win::Dialog, public Mt::IThreadX
{
public:
     Distribution()
     {
     }
     ~Distribution()
     {
     }
     wstring text;
     Mt::CriticalSection cs;
     Mt::ThreadPool pool;
     //_____________________________________________________________ Mt::IThreadX
     DWORD ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads);
protected:
     . . .
};

Step B

Edit the Distribution.cpp file.

Edite el archivo Distribution.cpp.

Distribution.cpp

. . .
void Distribution::Window_Open(Win::Event& e)
{
     //________________________________________________________ 1. Begin all threads in the pool
     pool.BeginAll(*this);
     //________________________________________________________ 2. Wait for all threads in the pool to finish
     pool.WaitForAll();
     tbxOutput.Text = text;
}

DWORD Distribution::ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads)
{
     const int total = 1000;
     const int first = (int)(threadIndex*(total/numThreads));
     const int last = (int)( (threadIndex == numThreads-1) ? total : (threadIndex+1)*(total/numThreads) );
     wchar_t tmp[256];
     _snwprintf_s(tmp, 256, _TRUNCATE, L"threadIndex=%d\r\tfirst=%d\t\tlast=%d\r\n", (int)threadIndex, first, last);
     cs.Enter();
     text += tmp;
     cs.Leave();
     return 0;
}

Problem 3

Create a Wintempla Dialog application called MatrixSec to multiply two matrices using several threads. In this case, each thread will compute a section of consecutive rows of the resulting matrix as shown. The main thread will start all threads in the pool, and then, it will wait for all threads to complete. Observe that the MatrixSec class implements one interface: Mt::IThreadX. Do not forget to check the WM_APP event in the main window.

Cree una aplicación de Dialogo usando Wintempla llamada MatrixSec para multiplicar dos matrices usando varias threads. En este caso, cada thread calculará una sección de renglones consecutivos en la matriz resultante como se muestra. La thread iniciará las thread en la pool, y entonces, esta esperará a que todas las threads terminen. Observe que la clase MatrixSec implementa una interface: Mt::IThreadX. No se olvide de seleccionar el evento WM_APP en la ventana principal.

MatrixSec.h

#pragma once //______________________________________ MatrixSec.h
#include "Resource.h"
#define WORK_ID 100

class MatrixSec: public Win::Dialog, public Mt::IThreadX
{
public:
     MatrixSec()
     {
          numThreads = 0;
     }
     ~MatrixSec()
     {
     }
     long rowCountA;
     long colCountA;
     long rowCountB;
     long colCountB;
     MATRIX matrixA;
     MATRIX matrixB;
     MATRIX matrixResult;
     //
     int numThreads;
     Sys::Stopwatch sw;
     Mt::ThreadPool pool;
     //___________________________________________________________________ Mt::IThreadX
     DWORD ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads);
     . . .
};

MatrixSec.cpp

. . .
void MatrixSec::Window_Open(Win::Event& e)
{
     long i, j;
     //___________________________________ 1. Create: matrixA, matrixB and matrixResult
     Math::Oper::CreateMatrix(matrixA, 2000, 800);
     Math::Oper::CreateMatrix(matrixB, 800, 300);
     Math::Oper::CreateMatrix(matrixResult, 2000, 300);
     rowCountA = (long)matrixA.size();
     colCountA = (long)matrixA[0].size();
     rowCountB =(long)matrixB.size();
     colCountB = (long)matrixB[0].size();
     //__________________________________ 2. Initialize A with random values
     for (i = 0; i < rowCountA; i++)
     {
          for (j = 0; j < colCountA; j++)
          {
               matrixA[i][j] = rand()/(double)RAND_MAX;
          }
     }
     //_________________________________ 3. Initialize B with random values
     for (i = 0; i < rowCountB; i++)
     {
          for (j = 0; j < colCountB; j++)
          {
               matrixB[i][j] = rand()/(double)RAND_MAX;
          }
     }
     //
     sw.Start();
     //_________________________________ 4. Start all threads in the pool
     numThreads = pool.BeginAll(*this);
     //_________________________________ 5. Display number of threads
     wchar_t text[64];
     _snwprintf_s(text, 64, _TRUNCATE, L"Threads: %d", numThreads);
     this->Text = text;
}

DWORD MatrixSec::ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads)
{
     size_t row = 0, col = 0, k = 0;
     double result = 0.0;
     const size_t first = Mt::WorkDistributionFirst(threadIndex, numThreads, matrixResult.size());
     const size_t last = Mt::WorkDistributionLast(threadIndex, numThreads, matrixResult.size());
     //____________________________________ 1. Compute resulting matrix from "first" to "last"
     for (row = first; row < last; row++)
     {
          for (col = 0; col < colCountB; col++)
          {
               result = 0.0;
               for (k = 0; k < colCountA; k++)
               {
                    result += matrixA[row][k] * matrixB[k][col];
               }
               //_______________________________ 2. Copy result to matrixResult[row][col]
               matrixResult[row][col] = result;
          }
     }
     //___________________________________ 3. Report completion to main thread
     ::PostMessage(hWnd, WM_APP, (WPARAM)threadIndex, (LPARAM)WORK_ID);
     return 0;
}

void MatrixSec::Window_App(Win::Event& e)
{
     if (e.lParam == WORK_ID)
     {
          const int threadIndex = (int)e.wParam;
          numThreads--;
          pool[threadIndex].WaitForExit();
          //pool.WaitForOne(threadIndex);
          //Sys::PrintLine(L"%d", pool.WaitForOne());
          if (numThreads == 0)
          {
               //pool.WaitForAll();
               this->Text = sw.GetMillisecondsText();
               //Sys::FileAssistant::CsvSave(L"C:\\selo\\MatrixSec.csv", matrixResult);
          }
     }
}

Problem 4

In teams of four students inspect and review the Mt::SuspendedThread class.

En equipos de cuatros estudiantes inspeccione y revise la clase Mt::SuspendedThread.

Problem 5

In teams of four students inspect and compare the Mt::ThreadPool class with the Mt::SuspendedThreadPool class.

En equipos de cuatros estudiantes inspeccione y compare la clase Mt::ThreadPool con la clase Mt::SuspendedThreadPool.

Problem 6

Create a Wintempla Dialog application called MatrixSuspended to multiply two matrices using a pool of suspended threads. The main difference between using Mt::ThreadPool and Mt::SuspendedThreadPool is that each worker thread begins execution by calling WakeUp instead of BeginThread.

Cree una aplicación de Dialogo usando Wintempla llamada MatrixSuspended para multiplicar dos matrices usando un conjunto de threads suspendidas. La diferencia principal entre usar Mt::ThreadPool y Mt::SuspendedThreadPool es que cada Worker Threads inicia ejecución llamando WakeUp en lugar de BeginThread.

MatrixSuspend.h

#pragma once //______________________________________ MatrixSuspend.h
#include "Resource.h"
#define WORK_ID 1

class MatrixSuspend: public Win::Dialog, public Mt::IThreadX
{
public:
     MatrixSuspend()
     {
     }
     ~MatrixSuspend()
     {
     }
     long rowCountA;
     long colCountA;
     long rowCountB;
     long colCountB;
     MATRIX matrixA;
     MATRIX matrixB;
     MATRIX matrixResult;
     //
     int numThreads;
     Sys::Stopwatch sw;
     Mt::SuspendedThreadPool pool;
     //___________________________________________________________________ Mt::IThreadX
     DWORD ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads);
protected:
     . . .
};

MatrixSuspend.cpp

. . .
void MatrixSuspend::Window_Open(Win::Event& e)
{
     long i, j;
     //___________________________________ 1. Create: matrixA, matrixB and matrixResult
     Math::Oper::CreateMatrix(matrixA, 2000, 800);
     Math::Oper::CreateMatrix(matrixB, 800, 300);
     Math::Oper::CreateMatrix(matrixResult, 2000, 300);
     rowCountA = (long)matrixA.size();
     colCountA = (long)matrixA[0].size();
     rowCountB =(long)matrixB.size();
     colCountB = (long)matrixB[0].size();
     //__________________________________ 2. Initialize A with random values
     for (i = 0; i < rowCountA; i++)
     {
          for (j = 0; j < colCountA; j++)
          {
               matrixA[i][j] = rand()/(double)RAND_MAX;
          }
     }
     //_________________________________ 3. Initialize B with random values
     for (i = 0; i < rowCountB; i++)
     {
          for (j = 0; j < colCountB; j++)
          {
               matrixB[i][j] = rand()/(double)RAND_MAX;
          }
     }
     //
     sw.Start();
     //_________________________________ 4. Start all threads in the pool
     numThreads = pool.WakeUpAll(*this);
     //_________________________________ 5. Display number of threads
     wchar_t text[64];
     _snwprintf_s(text, 64, _TRUNCATE, L"Threads: %d", numThreads);
     this->Text = text;
}

DWORD MatrixSuspend::ThreadFunc(Mt::BoolTs& cancel, size_t threadIndex, size_t numThreads)
{
     size_t row = 0, col = 0, k =0;
     double result = 0.0;
     const size_t first = Mt::WorkDistributionFirst(threadIndex, numThreads, matrixResult.size());
     const size_t last = Mt::WorkDistributionLast(threadIndex, numThreads, matrixResult.size());
     //____________________________________ 1. Compute resulting matrix from "first" to "last"
     for (row = first; row < last; row++)
     {
          for (col = 0; col < colCountB; col++)
          {
               result = 0.0;
               for (k = 0; k < colCountA; k++)
               {
                    result += matrixA[row][k] * matrixB[k][col];
               }
               //_______________________________ 2. Copy result to matrixResult[row][col]
               matrixResult[row][col] = result;
          }
     }
     //___________________________________ 3. Report completion to main thread
     ::PostMessage(hWnd, WM_APP, (WPARAM)threadIndex, (LPARAM)WORK_ID);
     return 0;
}

void MatrixSuspend::Window_App(Win::Event& e)
{
     if (e.lParam == WORK_ID)
     {
          const int threadIndex = (int)e.wParam;
          numThreads--;
          //pool[threadIndex].WaitForExit();
          pool.WaitForOne(threadIndex);
          if (numThreads == 0)
          {
               //pool.WaitForAll();
               this->Text = sw.GetMillisecondsText();
               //Sys::FileAssistant::CsvSave(L"C:\\selo\\MatrixSuspend.csv", matrixResult);
          }
     }
}

Thread Pool

Thread Pool

© Copyright 2000-2021 Wintempla selo. All Rights Reserved. Jul 22 2021. Home